package studio.raptor.databus.client;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.databus.command.ActiveMQObjectMessage;
import studio.raptor.databus.command.Event;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;

/**
 * 消费者对象
 */
public final class DatabusConsumer {

    private Connection connection;
    private Session session ;
    private MessageConsumer messageConsumer;
    private static Logger log = LoggerFactory.getLogger(DatabusConsumer.class);
    /**
     * 消费者构造方法-对包内其他类开放
     *
     */
    DatabusConsumer(){ }

    /**
     * 获取消息
     * <p>
     *
     * 若消费者配置中，指定ACK方式非自动，需要使用此方法接受消息，且需人工进行ACK消息，如下
     * <blockquote>
     * <pre>
     * ActiveMQObjectMessage msg = databusConsumer.receiveShouldClientACKMessage(3*1000);
     * doOther();
     * //手工确认消息
     * msg.acknowledge();
     * </pre>
     * </blockquote>
     * </p>
     *
     * @param receiveTimeout 接受消息超时时间
     * @return 若超过指定时间未收到消息或者receiveTimeout小于1L，返回NULL
     */
    public ActiveMQObjectMessage receiveShouldClientACKMessage(long receiveTimeout) throws JMSException {
        if( receiveTimeout <1L ){
            return null;
        }
        ActiveMQObjectMessage objectMessage = (ActiveMQObjectMessage)this.messageConsumer.receive(receiveTimeout);
        return objectMessage ;
    }

    /**
     * 获取消息。若Broker中无消息，会一直阻塞
     * @return
     * @throws JMSException
     */
    public Event receive() throws JMSException {
        Event event = null ;
        ActiveMQObjectMessage objectMessage =(ActiveMQObjectMessage) this.messageConsumer.receive();
        if (objectMessage != null) {
            event = objectMessage.getEvent();
        }
        return event;
    }

    /**
     * 获取消息。
     * @param receiveTimeout 获取消息超时时间，单位毫秒。非正数表示获取不到直接返回
     * @return 若超过指定接受时间依然无法接受到消息，返回NULL
     * @throws JMSException
     */
    public Event receive(long receiveTimeout) throws JMSException {
        if( receiveTimeout <1L ){
            return null;
        }
        Event event = null ;
        ActiveMQObjectMessage objectMessage =(ActiveMQObjectMessage) this.messageConsumer.receive(receiveTimeout);
        if (objectMessage != null) {
            event = objectMessage.getEvent();
        }
        return event;
    }

    /**
     * 获取消息，不设置获取消息的超时时间
     * @return 若无消息则返回NULL
     * @throws JMSException
     */
    public Event receiveNoWait() throws JMSException {
        Event event = null ;
        ActiveMQObjectMessage objectMessage =(ActiveMQObjectMessage) this.messageConsumer.receiveNoWait();
        if (objectMessage != null) {
            event = objectMessage.getEvent();
        }
        return event;
    }

    /**
     * 关闭消费者，关闭后不可再用此消费者者接受消息
     */
    public void close()  {
        try {
            if (this.messageConsumer != null) {
                this.messageConsumer.close();
            }
            if (this.session != null) {
                this.session.close();
            }
            if (this.connection != null) {
                this.connection.close();
            }
        } catch (JMSException e) {
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 将可以手动ACK的消息转换为消息事件对象
     * @param objectMessage 可手动ACK的消息对象
     * @return event 消息事件。入参为空时，直接返回NULL
     * @throws JMSException
     */
    public Event translate2Event(ActiveMQObjectMessage objectMessage) throws JMSException {
        if (objectMessage == null) {
            return null;
        }else {
            return objectMessage.getEvent() ;
        }
    }

    Connection getConnection() {
        return connection;
    }

    void setConnection(Connection connection) {
        this.connection = connection;
    }

    Session getSession() {
        return session;
    }

    void setSession(Session session) {
        this.session = session;
    }

    MessageConsumer getMessageConsumer() {
        return messageConsumer;
    }

    void setMessageConsumer(MessageConsumer messageConsumer) {
        this.messageConsumer = messageConsumer;
    }
}
